#include <cassert>
#include "miniutil/buf2type.h"
#include "miniutil/srpc.h"

/*
    c char
    s short
    i int32
    l int64
    b string/char*
    
    message format:
    length [4] following content length...
    sessid  [4]
    message type [2]  0 REQ 1 RESP
    format description length [4]
    format [format length]
	function name length
	function name[string]
    parameters
	....

	the function name is seem as the 1st parameter
*/

namespace miniutil { namespace frame {

int RpcMessage::ssessid = 0x1000000;

int RpcMessage::getSessionid()
{
    return ++ssessid;
}

//
//RpcMessage::RpcMessage(std::string& func, std::vector< ParaBasic*>& vparas)
//{
//    mfunction = func;
//    mvparas = vparas;
//    mmsgtype = REQU;
//}

RpcMessage::RpcMessage(const std::vector<std::string> &paras)
{
	mmsgtype = REQU;
	mfunction = paras[0];
	for(unsigned int i=0;i< paras.size(); i++)
	{
		RpcCallPara<std::string> *ps = new RpcCallPara<std::string>(paras[i]);
		mvparas.push_back(ps);
	}
}


RpcMessage::RpcMessage(int len, const char  *pbuf)
{
    mmsgtype  = RESP;
    fromBuffer(pbuf, len);
}

RpcMessage::~RpcMessage()
{
	for(unsigned int i=0; i< mvparas.size(); i++)
		delete mvparas[i];
	mvparas.clear();
}

int RpcMessage::toBuffer(char* pbuf, int len)
{
    int offset = 0;
    std::pair<string, int> parastus = getMsgStus(mvparas);
	int totallen = parastus.second + 14 + (int)parastus.first.length();
    SET_NET_INT_VALUE(pbuf, totallen -4);
    SET_NET_INT_VALUE(pbuf+4, getSessionid());
    SET_NET_SHORT_VALUE(pbuf+8, mmsgtype);
    SET_NET_INT_VALUE(pbuf+10, (int)parastus.first.length());
    
    memcpy(pbuf+ 14, parastus.first.data(), parastus.first.length());
    offset = 14+(int)parastus.first.length();
    for(unsigned int i=0; i< mvparas.size(); i++)
    {
        switch (mvparas[i]->getType())
        {
			case ParaBasic::PARA_CHAR:
                *(char*)(pbuf + offset) = dynamic_cast<RpcCallPara<char> *>(mvparas[i])->mvalue;
                offset += 1;
                break;
            case ParaBasic::PARA_SHORT:
                SET_NET_SHORT_VALUE(pbuf + offset, dynamic_cast<RpcCallPara<short> *>(mvparas[i])->mvalue);
                offset += 2;
                break;
            case ParaBasic::PARA_INT32:
                SET_NET_INT_VALUE(pbuf + offset, dynamic_cast<RpcCallPara<int> *>(mvparas[i])->mvalue);
                offset += 4;
                break;
            case ParaBasic::PARA_INT64:
                SET_NET_LL_VALUE(pbuf + offset, dynamic_cast<RpcCallPara<long long> *>(mvparas[i])->mvalue);
                offset += 8;
                break;
            case ParaBasic::PARA_STRING:
                {
                RpcCallPara<std::string> *pstring = dynamic_cast<RpcCallPara<std::string> *>(mvparas[i]);
                SET_NET_INT_VALUE(pbuf+offset, (int)pstring->mvalue.length());
                memcpy(pbuf + offset+4, pstring->mvalue.data(), pstring->mvalue.length());
                offset += (4+(int)pstring->mvalue.length());
				break;
				}
            case ParaBasic::PARA_BUFFER:
                {
                RpcCallPara<char*> *pparabuf = dynamic_cast<RpcCallPara<char*> *> (mvparas[i]);
                SET_NET_INT_VALUE(pbuf+offset, pparabuf->mlen);
                memcpy(pbuf + offset+4, pparabuf->mpvalue, pparabuf->mlen);
                offset += (4+pparabuf->mlen);
                break;
                }
            default:
                break;
       }
    }
    //assert(offset == totalmsglen);
    return offset;
}

int RpcMessage::fromBuffer(const char* pbufin, int len)
{
    mvparas.clear();
    
    msessid = GET_NET_INT_VALUE(pbufin+4);
    mmsgtype = GET_NET_SHORT_VALUE(pbufin+8);
    int formatlen;
    formatlen = GET_NET_INT_VALUE(pbufin+10);
    mformat = std::string(pbufin+14, formatlen);
    int offset = 14+formatlen;
    for(unsigned int i=0; i< mformat.length(); i++)
    {
        switch(mformat.at(i))
        {
            case 'c':
				{
                RpcCallPara<char> *pc = new RpcCallPara<char>((char)*(pbufin+offset));
                offset += 1;
                mvparas.push_back(pc);
                break;
				}
            case 's':
            {
                short vlue = GET_NET_SHORT_VALUE(pbufin+offset);
                RpcCallPara<short> *ps = new RpcCallPara<short>(vlue);
                mvparas.push_back(ps);
                offset += 2;
                break;
             }
            case 'i':
             {
                int vlue = GET_NET_INT_VALUE(pbufin+offset);
                RpcCallPara<int> *pi = new RpcCallPara<int>(vlue);
                mvparas.push_back(pi);
                offset += 4;
                break;
              }
           case 'l':
             {
                long long vlue = GET_NET_LL_VALUE(pbufin+offset);
                RpcCallPara<long long> *pl = new RpcCallPara<long long>(vlue);
                mvparas.push_back(pl);
                offset += 8;
              break;
               }
          case 'b':
             {
                int len = GET_NET_INT_VALUE(pbufin+offset);
                std::string content = std::string(pbufin+offset+4, len);
				if( i == 0)
					mfunction = content;
                RpcCallPara<std::string> *pb = new RpcCallPara<std::string>(content);
                mvparas.push_back(pb);
                offset += (4+len);
                break;
              }
       }
    }
    return 0;
}

int RpcMessage::getCreateMsgLen()
{
    std::pair<std::string, int> stus = getMsgStus(mvparas);
	return stus.second + 14 + (int)stus.first.length();
}

std::pair<std::string, int> RpcMessage::getMsgStus(const std::vector<ParaBasic*>& vparas)
{
    std::string format;
    int len = 0;
    for(unsigned int i=0; i< vparas.size(); i++)
    {
        switch (vparas[i]->getType())
        {
			case ParaBasic::PARA_CHAR:
				{
                format += "c";
                len += 1;
                break;
				}
            case ParaBasic::PARA_SHORT:
				{
                format += "s";
                len += 2;
                break;
				}
            case ParaBasic::PARA_INT32:
				{
                len += 4;
                format += "i";
                break;
				}
            case ParaBasic::PARA_INT64:
				{
                format += "l";
                len += 8;
                break;
				}
            case ParaBasic::PARA_STRING:
				{
                format += "b";
                RpcCallPara<std::string> *pstring = dynamic_cast<RpcCallPara<std::string> *> (mvparas[i]);
                len += 4;
                len += (int)pstring->mvalue.length();
				break;
				}
            case ParaBasic::PARA_BUFFER:
				{
                format += "b";
                RpcCallPara<char*> *pbuf = dynamic_cast<RpcCallPara<char*> *> (mvparas[i]);
                len += 4;
				len += pbuf->mlen;
				break;
				}
			default:
				break;
        }
    }
    return std::make_pair(format, len);
}
//-------------------------------------------------------------

RpcNode::RpcNode(miniutil::multinet::SocketMulplexService* pmul, miniutil::soaddr raddr)
:MultinetPeer(pmul, raddr)
{
}

RpcNode::RpcNode(miniutil::multinet::SocketMulplexService* pmul, miniutil::soaddr raddr, SOCKET soc)
:MultinetPeer(pmul, raddr, soc)
{
}

//int RpcNode::invoke(std::string func, int flag)
//{
//	ShareRpcMsg smsg = ShareRpcMsg::safenew(func);
//	return invoke(smsg, flag);
//}
//
//template <class T1>
//int RpcNode::invoke(std::string func, T1 para1, int flag)
//{
//	ShareRpcMsg smsg = ShareRpcMsg::safenew(func, para1);
//	return invoke(smsg, flag);
//}
//
//template <class T1, class T2>
//int RpcNode::invoke(std::string func, T1 para1, T2 para2, int flag)
//{
//	ShareRpcMsg smsg = ShareRpcMsg::safenew(func, para1, para2);
//	return invoke(smsg, flag);
//}
//
//template <class T1, class T2, class T3>
//int RpcNode::invoke(std::string func, T1 para1, T2 para2, T3 para3, int flag)
//{
//	ShareRpcMsg smsg = ShareRpcMsg::safenew(func, para1, para2, para3);
//	return invoke(smsg, flag);
//}
//
//template <class T1, class T2, class T3, class T4>
//int RpcNode::invoke(std::string func, T1 para1, T2 para2, T3 para3, T4 para4, int flag)
//{
//	ShareRpcMsg smsg = ShareRpcMsg::safenew(func, para1, para2, para3, para4);
//	return invoke(smsg, flag);
//}

#if 0
int RpcNode::invoke(std::string funcname, std::vector<ParaBasic*> vparas, int flag)
{
    RpcMessage msg(funcname, vparas);
    int totallen = msg.getCreateMsgLen();
    miniutil::data::shared_buffer sbuf =  miniutil::data::shared_buffer::safenew(miniutil::data::buffer(totallen));
    msg.toBuffer(sbuf->data(), sbuf->size());
    //send out
    if(flag & SYNC)
    {
        //wait for response
        //if timeout, return false
    }
    else if(flag & WAITRESP)
    {
        //add session to wait id.
    }
	return 0;
}
#endif

int RpcNode::invoke(ShareRpcMsg smsg, int flag)
{
	int ret;
    int totallen = smsg->getCreateMsgLen();
    miniutil::data::shared_buffer sbuf =  miniutil::data::shared_buffer::safenew(totallen);
    ret = smsg->toBuffer(sbuf->data(), (int)sbuf->size());
	assert(ret == totallen);
	sbuf->set_length(totallen);
    SendoutData(sbuf);
    //send out
    if(flag & SYNC)
    {
        while(true)
        {
            //mwaitingsession.insert(smsg->msessid);
            if(mvremoterespmsgs.find(smsg->msessid) != mvremoterespmsgs.end())
                break;
        }
        //wait for response
        //if timeout, return false
    }
    else if(flag & WAITRESP)
    {
        //mwaitingsession.insert(smsg->msessid);
        //add session to wait id.
    }
    return smsg->msessid;
}	

//int RpcNode::analyze(char* pbuf, int len)
//{
//    RpcMessage msg(pbuf, len);
//	if(msg.mmsgtype == RpcMessage::REQU)
//		mvremotereqmsgs.push_back(msg);
//	else
//		mvremoterespmsgs.insert(std::make_pair(msg.msessid, msg));
//    return 0;
//}

int RpcNode::working()
{
    int ret = MultinetPeer::working();
    if(ret == -1)
    {
        return -1;
    }
    ret = mmsgreader.Read(&mrecvbuf);
    //retrieve data from network.
    //when finish one message
    //parse it
    if(ret == 1) //finish
    {
        ShareRpcMsg smsg = ShareRpcMsg::safenew(mmsgreader.GetMsgBufLen(), mmsgreader.RefMsgBuf());
	    if(smsg->mmsgtype == RpcMessage::REQU)
		    mvremotereqmsgs.push_back(smsg);
	    else
		    mvremoterespmsgs.insert(std::make_pair(smsg->msessid, smsg));
        mmsgreader.Clear();
    }
    return 0;
}

//int RpcNode::getRemoteRpcResp(int sessid, ShareRpcMsg& msg)
//{
//    if(mwaitingsession.find(sessid) == mwaitingsession.end())
//        return -1;
//    msg = mvremoterespmsgs[sessid];
//    mwaitingsession.erase(sessid);
//    mvremoterespmsgs.erase(sessid);
//    return 0;
//}
//
//int RpcNode::getRemoteRpcReq(std::vector< ShareRpcMsg>& vmsgs)
//{
//    vmsgs = mvremotereqmsgs;
//    mvremotereqmsgs.clear();
//    return 0;
//}
//

RpcService::RpcService(unsigned short port)
:MultinetService<RpcNode>(port)
{
}

void RpcService::broadcast(ShareRpcMsg smsg)
{
    miniutil::auto_lock lk(&mtx);
    std::vector<RpcNode*>::iterator iter_vp;
    for(iter_vp = mvpeers.begin(); iter_vp != mvpeers.end(); iter_vp ++)
    {
        (*iter_vp)->invoke(smsg, false);
    }
}
    
void RpcService::checkNode(RpcNodeHandler *phandler)
{
    int ret;
    miniutil::auto_lock lk(&mtx);
    std::vector<RpcNode*>::iterator iter_vp;
    for(iter_vp = mvpeers.begin(); iter_vp != mvpeers.end(); iter_vp ++)
    {
        ret = phandler->operator()(*iter_vp);
        if(ret == -1)
        {
            delete *iter_vp;
            iter_vp = mvpeers.erase(iter_vp);
        }
    }
}



} }
